package quota

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	xecode "go-common/library/ecode"
	"go-common/library/log"
	"go-common/library/net/rpc/warden"
	"go-common/library/rate/limit/quota/api"
	"go-common/library/stat/metric"

	"go-common/library/rate/limit/quota/rate"
)

const (
	_retryTime = time.Millisecond * 500
	_infRetry  = 0
	_bucketLen = 5
)

var (
	_ripeDuration = time.Second * 5
	_metricQuota  = metric.NewCounterVec(&metric.CounterVecOpts{
		Namespace: "rate",
		Subsystem: "limit",
		Name:      "quota",
		Help:      "quota limit total.",
		Labels:    []string{"id"},
	})
)

type resource struct {
	// 资源id
	id string
	// 刷新时间(秒)
	refreshInterval int64
	// 获得的配额
	gets float64
	// 可以临时占用的配额
	reserve float64
	lm      *rate.Limiter
	// 主节点地址
	masterAddr string
	conn       api.QuotaClient
	closeFn    func() error
	closCh     chan bool
	updateCh   chan float64
	closed     bool
	// 接受限流的时间
	ripeTime time.Time
	// 是否是未注册的资源
	unknow bool
	// 请求统计
	requests [_bucketLen]int64
	// 使用统计
	used [_bucketLen]int64
}

var directCall func(masterAddr string) (client api.QuotaClient, closer func() error, err error)

func (l *resource) tickerproc() {
	tick := time.NewTicker(time.Second * time.Duration(l.refreshInterval))
	lastInterval := l.refreshInterval
	defer tick.Stop()
	for {
		select {
		case <-l.closCh:
			return
		case <-tick.C:
			newInterval := atomic.LoadInt64(&l.refreshInterval)
			if lastInterval != newInterval {
				tick.Stop()
				log.Info("quota: resource(%s) set new refresh interval: %ds old: %ds", l.id, newInterval, lastInterval)
				tick = time.NewTicker(time.Duration(newInterval) * time.Second)
				lastInterval = newInterval
			}
			select {
			case l.updateCh <- l.getWants():
			default:
			}
		}
	}
}

func (l *resource) syncerproc() {
	for {
		select {
		case <-l.closCh:
			return
		case wants := <-l.updateCh:
			_ = l.refreshResource(_infRetry, wants)
		}
	}
}

func (l *resource) refreshResource(retry int, wants float64) (err error) {
	var resp *api.GetCapacityResp
	var info string
	defer func() {
		if info == "" {
			return
		}
		log.Info("quota: time: %s %s", time.Now().Format(time.RFC3339), info)
	}()
	err = l.callRPC(retry, func() (err error) {
		resp, err = l.conn.GetCapacity(context.Background(), &api.GetCapacityReq{
			ClientID:   clientAddr(),
			ResourceID: l.id,
			Wants:      wants,
		}, warden.WithLogFlag(warden.LogFlagDisable))
		if err != nil {
			if xecode.EqualError(api.QuotaResourceErr, err) || xecode.EqualError(api.QuotaNotMasterErr, err) || xecode.EqualError(xecode.LimitExceed, err) {
				log.Info("quota: resource(%s) GetCapacity(client: %s wants: %v) err: %+v", l.id, clientAddr(), wants, err)
			} else {
				log.Error("quota: resource(%s) GetCapacity(client: %s wants: %v) err: %+v", l.id, clientAddr(), wants, err)
			}
		} else {
			limit := wants - resp.Gets - resp.Reserve
			if limit > 0 {
				_metricQuota.Add(limit*float64(l.refreshInterval), l.id)
			}
			info = fmt.Sprintf("%sresource(%s) GetCapacity(client: %s wants: %v) gets %v reserve: %v refresh: %v  current_idx: %v want_bucket: %v used_bucket: %v ", info, l.id, clientAddr(), wants, resp.Gets, resp.Reserve, resp.RefreshInterval, l.idx(), l.requests, l.used)
		}
		return
	})
	if err != nil {
		if xecode.EqualError(api.QuotaResourceErr, err) {
			l.unknow = true
		}
		return
	}
	l.unknow = false
	if wants != 0 && l.ripeTime.IsZero() {
		l.ripeTime = time.Now().Add(time.Second*time.Duration(l.refreshInterval) + _ripeDuration)
	}
	l.gets = resp.Gets
	l.reserve = resp.Reserve
	interval := resp.RefreshInterval
	if interval <= 0 {
		info = fmt.Sprintf("%s get wrong refresh interval, use default ", info)
		atomic.StoreInt64(&l.refreshInterval, 1)
	} else {
		atomic.StoreInt64(&l.refreshInterval, interval)
	}
	if !l.ripeTime.IsZero() && time.Now().After(l.ripeTime) {
		limit := rate.Limit(l.gets + l.reserve)
		burst := int(limit)
		if burst <= 0 {
			burst = 1
		}
		if limit < 0 {
			limit = 0
		}
		if l.lm == nil {
			info = fmt.Sprintf("%s set new limiter: %v burst: %v ", info, limit, burst)
			l.lm = rate.NewLimiter(limit, burst)
			return
		}
		if l.lm.Burst() != burst {
			info = fmt.Sprintf("%s update burst: %v -> %v ", info, l.lm.Burst(), burst)
			l.lm.SetBurst(burst)
		}
		if l.lm.Limit() != limit {
			info = fmt.Sprintf("%s update limiter: %v -> %v ", info, l.lm.Limit(), limit)
			l.lm.SetLimit(limit)
		}
	}
	return
}

func (l *resource) Close() {
	if l.closed {
		return
	}
	l.closed = true
	l.closCh <- true
	_ = l.callRPC(3, func() (err error) {
		if _, err = l.conn.ReleaseCapacity(context.Background(), &api.ReleaseCapacityReq{
			ClientID:   clientAddr(),
			ResourceID: l.id,
		}); err != nil {
			return
		}
		fn := l.closeFn
		if fn != nil {
			_ = fn()
		}
		return
	})
}

func getMasterAddr(id string) (host string, err error) {
	req := &api.GetMasterAddrReq{ResourceID: id}
	reply, err := defaultClient.GetMasterAddr(context.Background(), req, warden.WithLogFlag(warden.LogFlagDisable))
	if err != nil {
		return
	}
	host = reply.MasterAddr
	return
}

func (l *resource) callRPC(maxRetry int, f func() (err error)) (err error) {
	var retry = 0
	if defaultClient == nil {
		panic("must init quota")
	}
	for {
		if maxRetry != _infRetry && retry >= maxRetry {
			return
		}
		if l.conn == nil {
			if l.masterAddr, err = getMasterAddr(l.id); err != nil || l.masterAddr == "" {
				log.Error("quota: resource(%s) get master err: %+v resp: %s", l.id, err, l.masterAddr)
				retry++
				time.Sleep(_retryTime)
				continue
			}
			log.Info("quota: resource(%s) get master addr: %s", l.id, l.masterAddr)
			if l.conn, l.closeFn, err = directCall(l.masterAddr); err != nil {
				retry++
				time.Sleep(_retryTime)
				continue
			}
		}
		err = f()
		switch xecode.Cause(err).Code() {
		case api.QuotaResourceErr.Code():
			return
		case api.QuotaNotMasterErr.Code():
			log.Warn("quota: resource(%s) host(%s) is not master retrying...", l.id, l.masterAddr)
			if l.closeFn != nil {
				_ = l.closeFn()
			}
			l.conn = nil
			l.closeFn = nil
			// 防止server节点不一致 切换过于频繁
			time.Sleep(_retryTime)
		case int(xecode.OK):
			return
		case xecode.LimitExceed.Code():
			return
		default:
			log.Error("quota: resource(%s) unexpected error(%+v) retrying...", l.id, err)
			if l.closeFn != nil {
				_ = l.closeFn()
			}
			l.closeFn = nil
			l.conn = nil
			retry++
			time.Sleep(_retryTime)
		}
	}
}

func directWardenCall(masterAddr string) (client api.QuotaClient, closer func() error, err error) {
	target := fmt.Sprintf("direct://default/%s", masterAddr)
	cc, err := warden.NewConn(target)
	if err != nil {
		log.Error("quota: dial %s err: %+v", target, err)
		return
	}
	closer = cc.Close
	client = api.NewQuotaClient(cc)
	return
}

func (l *resource) getWants() float64 {
	idx := l.idx()
	next := l.nextIdx(idx)
	atomic.StoreInt64(&l.requests[next], 0)
	atomic.StoreInt64(&l.used[next], 0)
	last := float64(atomic.LoadInt64(&l.requests[l.lastIdx(idx)]))
	return last / float64(l.refreshInterval)
}

// lastIdx 获取本次请求计数的下标值
func (l *resource) idx() int {
	return int(time.Now().Unix()/l.refreshInterval) % _bucketLen
}

// lastIdx 获取上一次请求计数的下标值
func (l *resource) lastIdx(idx int) int {
	idx--
	if idx < 0 {
		idx = _bucketLen + idx
	}
	return idx
}

func (l *resource) nextIdx(idx int) int {
	idx++
	return idx % _bucketLen
}
